/*
 * Copyright (c) 2015. 速位中国
 */

package com.netty;


import com.kit.NamedThreadFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Scope;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;


@Component
public class ServerService {

    private final static Logger logger = LoggerFactory.getLogger(ServerService.class);

    /**
     * 网络连接读取数据超时次数
     */
    private int nettyReadTimeoutTimes;
    /**
     * 当前服务器端口编号
     */
    private int port;
    /**
     * 服务器连接超时时间
     */
    private int nettyConnectTimeout;
    /**
     *
     */
    private int nettyReaderIdleTime;
    /**
     * linux  epoll
     */
    private boolean epoll;

    /**
     * 网络工作线程数量
     * [CPU+1,CPU*2]
     */
    private static final int PROCESSORS = Runtime.getRuntime().availableProcessors() * 2;

    /**
     * 业务处理线程池
     * <p/>
     * CPU*2
     */
    private static final EventExecutorGroup LOGIC_EXECUTORS = new
            DefaultEventExecutorGroup(PROCESSORS, new NamedThreadFactory("业务处理线程"));

    /**
     *  spring 环境变量
     */
    @Resource
    private Environment environment;


    /***
     *
     * 初始化网络基础配置
     * @throws IOException
     */
    @PostConstruct
    private void init() throws IOException {
        nettyReadTimeoutTimes = Integer.parseInt(environment.getProperty("netty.read.timeout.times"));
        port = Integer.parseInt(environment.getProperty("netty.port"));
        nettyConnectTimeout = Integer.parseInt(environment.getProperty("netty.connect.timeout"));
        nettyReaderIdleTime = Integer.parseInt(environment.getProperty("netty.readerIdle.time"));
        epoll = Boolean.parseBoolean(environment.getProperty("netty.net.epoll", "false"));
    }


    /**
     * 业务处理
     * @return
     */
    @Bean
    @Scope("prototype")
    public ServerHandler serverHandler() {
        return new ServerHandler(nettyReadTimeoutTimes);
    }

    /**
     * 解码器
     * @return
     */
    @Bean
    @Scope("prototype")
    public ServerDecoder serverDecoder() {
        return new ServerDecoder(0, 4, 0, 4);
    }

    /**
     * 心跳处理
     * @return
     */
    @Bean
    @Scope("prototype")
    public IdleStateHandler idleStateHandler() {
        return new IdleStateHandler(nettyReaderIdleTime, 0, 0, TimeUnit.SECONDS);
    }


    /**
     * 启动网络服务
     *
     * @throws Exception
     */
    public void startServer() {
        if (logger.isDebugEnabled()) {
            logger.debug(epoll ? "use epoll net work......" : "use nio net work......");
        }
        if (epoll) {
            try {
                runNativeEpoll();
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else {
            try {
                runNio();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * run netty server  with nio
     */
    private void runNio() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1, new NamedThreadFactory("BOSS"));//
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(PROCESSORS, new NamedThreadFactory("I/O线程"));//
        this.run0(bossGroup, workerGroup, NioServerSocketChannel.class);
    }


    /**
     * 基于jni，  c语言实现的native Epoll
     * #NioEventLoopGroup → EpollEventLoopGroup   ----实现了 EventLoopGroup 接口
     * #NioEventLoop → EpollEventLoop
     * #NioServerSocketChannel → EpollServerSocketChannel
     * #NioSocketChannel → EpollSocketChannel        ---- 实现了SocketChannel接口
     * <p/>
     * ## RHEL/CentOS/Fedora:
     * #sudo yum install autoconf automake libtool glibc-devel.i686 glibc-devel libgcc.i686 make
     * ## Debian/Ubuntu:
     * #sudo apt-get install autoconf automake libtool make gcc-multilib
     *
     * @throws Exception
     */
    private void runNativeEpoll() throws Exception {
        EpollEventLoopGroup bossGroup = new EpollEventLoopGroup(1, new NamedThreadFactory("BOSS"));
        EpollEventLoopGroup workerGroup = new EpollEventLoopGroup(PROCESSORS, new NamedThreadFactory("I/O线程"));
        this.run0(bossGroup, workerGroup, EpollServerSocketChannel.class);
    }


    /**
     * 启动网络服务
     *
     * @param bossGroup
     * @param workerGroup
     */
    private void run0(EventLoopGroup bossGroup, EventLoopGroup workerGroup, Class<? extends ServerChannel> channelClass) {
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(channelClass)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyConnectTimeout)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.TCP_NODELAY, true)

                    // TODO: 11/30/15 接收 发送 缓冲区大小 根据实际消息类型来设置
                    .option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(128, 1024, 65536))
                    .option(ChannelOption.SO_SNDBUF, 512)
                    .option(ChannelOption.SO_RCVBUF, 512)
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)//// TODO: 11/30/15 使用内存池
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);


            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                protected void initChannel(SocketChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    //I/O work thread
                    pipeline.addLast("heart", idleStateHandler());
                    pipeline.addLast("decode", serverDecoder());
                    // add logic executors
                    pipeline.addLast(LOGIC_EXECUTORS, "handler", serverHandler());

                }
            });


            ChannelFuture future = bootstrap.bind(new InetSocketAddress(port)).sync();//bind port
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        logger.info("port：" + port + " service started... ...");
                    }
                }
            });
            // Wait until the server socket is closed.
            // In this server, this does not happen, but you can do that to gracefully
            // shut down your server.
            future.channel().closeFuture().sync();

        } catch (Exception e) {
            logger.error(e.getMessage());
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }


}
